package cn.zwx.canal.tools.kafka;

import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * @description: 发送消息类
 * @projectName:learn
 * @see:cn.zwx.canal.tools.kafka
 * @author:zhangwenxue
 * @createTime:2021/1/12 14:44
 * @version:1.0
 */
@Component
public class KafKaRegisterProducer  {

    private Logger logger = LoggerFactory.getLogger(KafKaRegisterProducer.class);

    @Autowired
    @Qualifier("registerTemplate")
    private KafkaTemplate<String, String> kafkaTemplate ;

    private final static String topic = "CANAL_LISTENER_MYSQL_QUEUE";

    /**
     * @description 发送消息方法
     * @param message 发送消息json
     * @return void
     * @author zhangwenxue
     * @createTime 2021/1/12 15:09
     **/
    public void sendMessage(String message) {
        try {
            ListenableFuture<SendResult<String, String>> result = kafkaTemplate.send(topic, message);
            logger.info("发现消息成功.,发现结果:{}", JSON.toJSONString(result));
        }catch (Exception e){
            logger.error("发送kafka消息异常,原因:{}",e.getMessage(),e);
        }
    }

}
